package com.shujia.flink.state

import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object Demo1StateWordCount {
  def main(args: Array[String]): Unit = {


    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    env.setParallelism(1)

    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)

    val kvDS: DataStream[(String, Int)] = linesDS.flatMap(_.split(",")).map((_, 1))


    /**
      * sum reduce这种聚合函数，内会为每一个key保存一个状态（之前的统计结果）
      *
      */
    kvDS
      .keyBy(_._1)
      .process(new KeyedProcessFunction[String, (String, Int), (String, Int)] {

        /**
          * 使用scala普通变量保存计算结果（状态），如果任务失败重启状态会丢失
          *
          */
        var count = 0

        /**
          * 每一条数据才处理一次
          *
          * @param value : 一行数据
          * @param ctx   ： 上下文对象
          * @param out   ： 用于将数据发送到下游
          */
        override def processElement(value: (String, Int), ctx: KeyedProcessFunction[String, (String, Int), (String, Int)]#Context, out: Collector[(String, Int)]): Unit = {

          //统计单词的数量
          count += value._2

          //将数据发送到下游
          out.collect((value._1, count))
        }
      })
      .print()


    env.execute()

  }
}
